# akshare_finance/pipeline.py
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Dict, Tuple, Any, Optional
from akshare_finance.models import Company, Sector
from akshare_finance.fetcher import FinancialDataFetcher
from akshare_finance.analyzer import FinancialDataAnalyzer
from akshare_finance.dupont import DupontAnalyzer
import pandas as pd


def fetch_and_analyze(
    fetcher: FinancialDataFetcher,
    analyzer: FinancialDataAnalyzer,
    dupont_analyzer: DupontAnalyzer,
    company: Company,
) -> Tuple[Optional[pd.DataFrame], Optional[pd.DataFrame], Optional[pd.DataFrame], Optional[pd.DataFrame]]:
    main_business = fetcher.get_main_business(company.symbol)
    fin = fetcher.get_financial_abstract(company.symbol)
    analyzed = analyzer.analyze_financial_abstract(fin)
    # 杜邦分析（基础 + 改进）
    dupont_basic = dupont_analyzer.compute_basic_dupont(company.symbol)
    dupont_advanced = dupont_analyzer.compute_advanced_dupont(company.symbol)
    return analyzed, main_business, dupont_basic, dupont_advanced


def collect_all_sectors_data(
    sectors_map: Dict[str, list],
    workers: int,
    fetcher: FinancialDataFetcher,
    analyzer: FinancialDataAnalyzer,
    dupont_analyzer: DupontAnalyzer,
    reporter=None,
):
    """
    返回 all_sectors_data:
    { sector_name: (Sector对象, {公司名: (财务分析DF, 主营DF)}, 板块统计DF) }
    """
    all_sectors_data: Dict[
        str, Tuple[Sector, Dict[str, Tuple[Optional[pd.DataFrame], Optional[pd.DataFrame], Optional[pd.DataFrame], Optional[pd.DataFrame]]], Optional[pd.DataFrame]]
    ] = {}

    for sector_name, companies_list in sectors_map.items():
        sector = Sector(name=sector_name)
        sector.add_companies(companies_list)
        if reporter:
            reporter.print_sector_header(sector_name)

        companies_data = {}
        with ThreadPoolExecutor(max_workers=workers) as ex:
            futures = {
                ex.submit(fetch_and_analyze, fetcher, analyzer, dupont_analyzer, c): c
                for c in sector.companies
            }
            for fut in as_completed(futures):
                comp = futures[fut]
                try:
                    analyzed, main_business, dupont_basic, dupont_advanced = (
                        fut.result()
                    )
                except Exception as e:
                    # 让调用方自己记录日志；这里不打印，避免重复
                    analyzed, main_business, dupont_basic, dupont_advanced = (
                        None,
                        None,
                        None,
                        None,
                    )
                companies_data[comp.name] = (
                    analyzed,
                    main_business,
                    dupont_basic,
                    dupont_advanced,
                )
                if reporter:
                    reporter.print_company_report(
                        comp.name,
                        comp.symbol,
                        analyzed,
                        main_business,
                        dupont_basic,
                        dupont_advanced,
                    )

        sector_stats = analyzer.calculate_sector_statistics(
            {k: v[0] for k, v in companies_data.items()}
        )
        all_sectors_data[sector_name] = (sector, companies_data, sector_stats)

        # 板块统计打印
        if reporter:
            try:
                reporter.print_sector_statistics(sector_name, sector_stats)
            except AttributeError:
                # 老版本 reporter 没有这个方法就忽略
                pass

    return all_sectors_data
